/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.hylanda.processor;

import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.sql.Connection;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.apache.nifi.annotation.behavior.ReadsAttribute;
import org.apache.nifi.annotation.behavior.ReadsAttributes;
import org.apache.nifi.annotation.behavior.WritesAttribute;
import org.apache.nifi.annotation.behavior.WritesAttributes;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.SeeAlso;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnRemoved;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessorInitializationContext;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.InputStreamCallback;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.stream.io.StreamUtils;
import org.apache.nifi.util.StringUtils;

import com.alibaba.fastjson.JSONObject;
import com.hylanda.processor.database.util.Constant;
import com.hylanda.processor.database.util.SqlServerHelper;

/**
 * 
 * @author linda
 * @version created at：2017年11月24日
 * 
 */
@Tags({ "valley", "quadrant" })
@CapabilityDescription("接收上层传递的数据，写入Azure上的platform数据库的M_MagicQuadrant；" + "在配置界面中配置展现参数，写入M_MagicQuadrant_cfg表；"
		+ "通过平台提供的接口获取processor_id，并写入两张相关表。")
@SeeAlso({})
@ReadsAttributes({ @ReadsAttribute(attribute = "", description = "") })
@WritesAttributes({ @WritesAttribute(attribute = "", description = "") })
public class MagicQuadrantProcessor extends AbstractProcessor {
	static final String YES = "是";
	static final String NO = "否";
	public static final PropertyDescriptor CFG_DESCRIPTOR = new PropertyDescriptor.Builder().name("cfg")
			.displayName("模型显示配置").required(true).allowableValues("        ").defaultValue("        ").build();
	public static final PropertyDescriptor TITLE_PROPERTY = new PropertyDescriptor.Builder().name("model-title")
			.displayName("◆            四象限显示标题").description("四象限模型中需要显示的标题").required(false)
			.addValidator(StandardValidators.NON_EMPTY_VALIDATOR).defaultValue("魔力四象限").build();

	public static final PropertyDescriptor X_AXIS_PROPERTY = new PropertyDescriptor.Builder()
			.name("x-axis-displayname").displayName("◆            X轴显示名称").description("四象限模型中X轴显示名称")
			.required(false).addValidator(StandardValidators.NON_EMPTY_VALIDATOR).defaultValue("前瞻性").build();

	public static final PropertyDescriptor Y_AXIS_PROPERTY = new PropertyDescriptor.Builder()
			.name("y-axis-displayname").displayName("◆            Y轴显示名称").description("四象限模型中Y轴显示名称")
			.required(false).addValidator(StandardValidators.NON_EMPTY_VALIDATOR).defaultValue("执行力").build();

	public static final PropertyDescriptor FIRST_QUADRANT_PROPERTY = new PropertyDescriptor.Builder()
			.name("first-quadrant-displayname").displayName("◆            第一象限显示名称").description("四象限模型中第一象限显示名称")
			.required(false).addValidator(StandardValidators.NON_EMPTY_VALIDATOR).defaultValue("领导者").build();

	public static final PropertyDescriptor SECOND_QUADRANT_PROPERTY = new PropertyDescriptor.Builder()
			.name("second-quadrant-displayname").displayName("◆            第二象限显示名称")
			.description("四象限模型中第二象限显示名称").required(false).defaultValue("挑战者")
			.addValidator(StandardValidators.NON_EMPTY_VALIDATOR).build();

	public static final PropertyDescriptor THIRD_QUADRANT_PROPERTY = new PropertyDescriptor.Builder()
			.name("third-quadrant-displayname").displayName("◆            第三象限显示名称").description("四象限模型中第三象限显示名称")
			.required(false).addValidator(StandardValidators.NON_EMPTY_VALIDATOR).defaultValue("参与者").build();

	public static final PropertyDescriptor FOURTH_QUADRANT_PROPERTY = new PropertyDescriptor.Builder()
			.name("fourth-quadrant-displayname").displayName("◆            第四象限显示名称")
			.description("四象限模型中第四象限显示名称").required(false).defaultValue("远见者")
			.addValidator(StandardValidators.NON_EMPTY_VALIDATOR).build();

	public static final PropertyDescriptor DATA_DESCRIPTOR = new PropertyDescriptor.Builder().name("data")
			.displayName("模型数据配置").required(true).allowableValues("        ").defaultValue("        ").build();

	public static final PropertyDescriptor OBJECT_PROPERTY = new PropertyDescriptor.Builder()
			.name("object-analyzed").displayName("◆            对象").description("待分析对象字段名称").required(false)
			.addValidator(StandardValidators.NON_EMPTY_VALIDATOR).build();

	public static final PropertyDescriptor X_AXIS_DATA_PROPERTY = new PropertyDescriptor.Builder()
			.name("x-axis-data").displayName("◆            X轴数据").description("X轴数据字段名称").required(false)
			.addValidator(StandardValidators.NON_EMPTY_VALIDATOR).build();

	public static final PropertyDescriptor Y_AXIS_DATA_PROPERTY = new PropertyDescriptor.Builder()
			.name("y-axis-data").displayName("◆            Y轴数据").description("Y轴数据字段名称").required(false)
			.addValidator(StandardValidators.NON_EMPTY_VALIDATOR).build();

	public static final PropertyDescriptor RADIX_PROPERTY = new PropertyDescriptor.Builder().name("radix")
			.displayName("◆            基数").description("基数字段名称（针对对象的统计数值）").required(false)
			.addValidator(StandardValidators.NON_EMPTY_VALIDATOR).build();

	public static final PropertyDescriptor CONTROL_DESCRIPTOR = new PropertyDescriptor.Builder().name("control")
			.displayName("模型控制").required(true).allowableValues("        ").defaultValue("        ").build();

	public static final PropertyDescriptor CONTROL_VALUES = new PropertyDescriptor.Builder()
			.name("control-values").displayName("◆            启动时是否清空已有模型数据").description("启动时默认不清空已有模型数据")
			.required(false).allowableValues(YES, NO).defaultValue(NO).build();

	static final Relationship REL_SUCCESS = new Relationship.Builder().name("成功")
			.description("成功的更新了四象限模型显示配置，输入数据成功保存").build();

	static final Relationship REL_FAILURE = new Relationship.Builder().name("失败")
			.description("未成功更新四象限模型显示配置，输入数据未成功保存，输入数据非法").build();

	private List<PropertyDescriptor> descriptors;

	private Set<Relationship> relationships;

	@Override
	protected void init(final ProcessorInitializationContext context) {
		final List<PropertyDescriptor> descriptors = new ArrayList<PropertyDescriptor>();
		descriptors.add(CFG_DESCRIPTOR);
		descriptors.add(TITLE_PROPERTY);
		descriptors.add(X_AXIS_PROPERTY);
		descriptors.add(Y_AXIS_PROPERTY);
		descriptors.add(FIRST_QUADRANT_PROPERTY);
		descriptors.add(SECOND_QUADRANT_PROPERTY);
		descriptors.add(THIRD_QUADRANT_PROPERTY);
		descriptors.add(FOURTH_QUADRANT_PROPERTY);
		descriptors.add(DATA_DESCRIPTOR);
		descriptors.add(OBJECT_PROPERTY);
		descriptors.add(X_AXIS_DATA_PROPERTY);
		descriptors.add(Y_AXIS_DATA_PROPERTY);
		descriptors.add(RADIX_PROPERTY);
		descriptors.add(CONTROL_DESCRIPTOR);
		descriptors.add(CONTROL_VALUES);

		this.descriptors = Collections.unmodifiableList(descriptors);

		final Set<Relationship> relationships = new HashSet<Relationship>();
		relationships.add(REL_SUCCESS);
		relationships.add(REL_FAILURE);
		this.relationships = Collections.unmodifiableSet(relationships);
	}

	@Override
	public Set<Relationship> getRelationships() {
		return this.relationships;
	}

	@Override
	public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
		return descriptors;
	}

	@OnScheduled
	public void onScheduled(final ProcessContext context) {
		// 是否清空已有数据
		final boolean empty = context.getProperty(CONTROL_VALUES).getValue().equals(YES);
		try (final Connection conn = (Connection) SqlServerHelper.connectDb(Constant.hostname,
				Constant.databaseName, Constant.username, Constant.password);
				final Statement stmt = SqlServerHelper.getStatement(conn)) {
			// 清空旧数据
			if (empty) {
				stmt.executeUpdate("update M_MagicQuadrant set b_flag=0 where processor_id='" + getIdentifier()
						+ "'");
			}
		} catch (Exception e) {
			getLogger().info("清空旧数据失败，processor_id=" + getIdentifier());
			e.printStackTrace();
		}
	}

	@OnRemoved
	public void onRemoved() {
		try (final Connection conn = (Connection) SqlServerHelper.connectDb(Constant.hostname,
				Constant.databaseName, Constant.username, Constant.password);
				final Statement stmt = SqlServerHelper.getStatement(conn)) {
			conn.setAutoCommit(false);
			stmt.addBatch("delete from M_MagicQuadrant_cfg where processor_id='" + getIdentifier() + "'");
			stmt.addBatch("update M_MagicQuadrant set b_flag=0 where processor_id='" + getIdentifier() + "'");
			stmt.executeBatch();
			conn.commit();
		} catch (Exception e) {
			getLogger().info("删除魔力四象限数据失败，processor_id=" + getIdentifier());
			e.printStackTrace();
		}
	}

	@Override
	public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
		FlowFile flowFile = session.get();
		if (flowFile == null) {
			return;
		}
		final ComponentLog logger = getLogger();

		// 解析数据信息
		final String objectAnalyzed = context.getProperty(OBJECT_PROPERTY).getValue();
		final String xAxisData = context.getProperty(X_AXIS_DATA_PROPERTY).getValue();
		final String yAxisData = context.getProperty(Y_AXIS_DATA_PROPERTY).getValue();
		final String radix = context.getProperty(RADIX_PROPERTY).getValue();

		// 读取flowFile内容
		final byte[] buffer = new byte[(int) flowFile.getSize()];
		session.read(flowFile, new InputStreamCallback() {
			@Override
			public void process(final InputStream in) throws IOException {
				StreamUtils.fillBuffer(in, buffer);
			}
		});

		// 解析输入为json数组
		final JSONObject obj;
		try {
			obj = JSONObject.parseObject(new String(buffer, StandardCharsets.UTF_8));
		} catch (Exception e) {
			logger.error("input flowfile is not json");
			session.transfer(flowFile, REL_FAILURE);
			return;
		}
		final String objectValue = obj.getString(objectAnalyzed);
		final String xAxisValue = obj.getString(xAxisData);
		final String yAxisValue = obj.getString(yAxisData);
		final String radixValue = obj.getString(radix);

		// 保存信息
		try {
			this.saveCfg(context, objectValue, xAxisValue, yAxisValue, radixValue);
		} catch (Exception e1) {
			logger.error("saveToDB error {}", new Object[] { e1.getMessage() });
			session.transfer(flowFile, REL_FAILURE);
			return;
		}
		session.transfer(flowFile, REL_SUCCESS);
	}

	/**
	 * 保存显示配置
	 * 
	 * @param context
	 * @throws Exception
	 */
	private void saveCfg(ProcessContext context, String objectValue, String xAxisValue, String yAxisValue,
			String radixValue) throws Exception {
		final ComponentLog logger = getLogger();
		// 添加配置信息
		final String title = context.getProperty(TITLE_PROPERTY).getValue();
		final String xAxisDisplay = context.getProperty(X_AXIS_PROPERTY).getValue();
		final String yAxisDisplay = context.getProperty(Y_AXIS_PROPERTY).getValue();
		final String _1st = context.getProperty(FIRST_QUADRANT_PROPERTY).getValue();
		final String _2st = context.getProperty(SECOND_QUADRANT_PROPERTY).getValue();
		final String _3st = context.getProperty(THIRD_QUADRANT_PROPERTY).getValue();
		final String _4st = context.getProperty(FOURTH_QUADRANT_PROPERTY).getValue();
		final StringBuilder sqlAccumulator1 = new StringBuilder();
		sqlAccumulator1.append("if not exists (select 1 from M_MagicQuadrant_cfg where processor_id = '")
				.append(getIdentifier()).append("' )");
		sqlAccumulator1
				.append("insert into M_MagicQuadrant_cfg(processor_id, title,Y,X,[1st],[2st],[3st],[4st]) values('")
				.append(getIdentifier()).append("', '").append(title).append("', '").append(yAxisDisplay)
				.append("', '").append(xAxisDisplay).append("', '").append(_1st).append("', '").append(_2st)
				.append("', '").append(_3st).append("', '").append(_4st).append("') ");
		sqlAccumulator1.append("else update M_MagicQuadrant_cfg set title = '").append(title).append("',Y='")
				.append(yAxisDisplay).append("',X='").append(xAxisDisplay).append("',[1st]='").append(_1st)
				.append("',[2st]='").append(_2st).append("',[3st]='").append(_3st).append("',[4st]='")
				.append(_4st).append("' where processor_id = '").append(getIdentifier()).append("'");

		final StringBuilder sqlAccumulator = new StringBuilder();
		// 添加数据信息
		if (!StringUtils.isEmpty(objectValue) && !StringUtils.isEmpty(xAxisValue)
				&& !StringUtils.isEmpty(yAxisValue)) {
			try {
				Long xAxisL = Long.parseLong(xAxisValue);
				Long yAxisL = Long.parseLong(yAxisValue);
				boolean withRadix = !StringUtils.isEmpty(radixValue);
				sqlAccumulator.append("insert into M_MagicQuadrant(processor_id,legend,x_axis,y_axis");
				if (withRadix) {
					sqlAccumulator.append(",radial");
				}
				sqlAccumulator.append(") values('").append(getIdentifier()).append("','").append(objectValue)
						.append("',").append(xAxisL).append(",").append(yAxisL);
				if (withRadix) {
					sqlAccumulator.append(",").append(Long.parseLong(radixValue));
				}
				sqlAccumulator.append(")");
			} catch (Exception e) {
				sqlAccumulator.setLength(0);
				logger.error("saveDataToDB M_MagicQuadrant error {}", new Object[] { e.getMessage() });
			}
		}

		try (final Connection con = (Connection) SqlServerHelper.connectDb(Constant.hostname,
				Constant.databaseName, Constant.username, Constant.password);
				final Statement stmt = SqlServerHelper.getStatement(con)) {
			stmt.executeUpdate(sqlAccumulator1.toString());
			sqlAccumulator1.setLength(0);
			if (sqlAccumulator.length() > 0) {
				stmt.executeUpdate(sqlAccumulator.toString());
				sqlAccumulator.setLength(0);
			}
		} catch (Exception e) {
			throw e;
		}
	}
}
